/*
 * Copyright (c) 2011 Yoyo Systems. All rights reserved.
 *
 * $Id$
 */
package com.ebo.storage.redis;

import java.util.Arrays;
import java.util.SortedMap;
import java.util.TreeMap;

import redis.clients.util.Hashing;
import redis.clients.util.MurmurHash;

public class ShardingAlgorithm {
	
	private final static int DEFAULT_VIRTUAL_SHARDS_PER_SHARD = 40;
    private final TreeMap<Long, ShardedRedisPool> nodes = new TreeMap<Long, ShardedRedisPool>();
    private int virtualShardsPerShard;
    
    private final Hashing algo = new MurmurHash();
    
    private ShardedRedisPool[] pools;

    public ShardingAlgorithm(ShardedRedisPool[] pools) {
        this(pools, DEFAULT_VIRTUAL_SHARDS_PER_SHARD);
    }

    public ShardingAlgorithm(ShardedRedisPool[] pools, int virtualShardsPerShard) {

    	this.virtualShardsPerShard = virtualShardsPerShard;
        this.pools = pools;
        initialize(pools);
        
    }

    private void initialize(ShardedRedisPool[] pools) {
    	 for (ShardedRedisPool pool : pools) {
             add(pool);
         }
	}

	private void add(ShardedRedisPool pool) {
        StringBuilder sb = new StringBuilder("VS-");
        int len = sb.length();
        for (int idx = 0; idx < virtualShardsPerShard; ++idx) {
            sb.setLength(len);
            sb.append(idx);
            sb.append("-");
            sb.append(pool.getShardName());
            long hashCode = hash(sb.toString());
            nodes.put(hashCode, pool);
        }
    }

    @SuppressWarnings("unused")
    private void remove(ShardedRedisPool pool) {
        long hashCode = hash(pool.getShardName());
        if (!nodes.get(hashCode).equals(pool)) {
            throw new IllegalArgumentException("Invalid pool in hash circle");
        } else {
            nodes.remove(hashCode);
        }
    }

//    public static void main(String[] args) {
//		
//	}
    
    
    public ShardedRedisPool shard(byte[] key) {
    	int roleId = RedisStorageKey.getHash(key);
    	return this.shard(roleId);
    }
    
    private ShardedRedisPool shard(int roleId) {
    	if (nodes.isEmpty()) {
            throw new IllegalArgumentException("No pool exist");
         }
    	 
    	 SortedMap<Long, ShardedRedisPool> tail = nodes.tailMap(algo.hash(String.valueOf(roleId)));
         if (tail.isEmpty()) {
             return nodes.get(nodes.firstKey());
         }
         return tail.get(tail.firstKey());
    	
    }
    
    public ShardedRedisPool[] getPools() {
		return pools;
	}

	public int getVirtualShardsPerShard() {
		return virtualShardsPerShard;
	}

	protected ShardedRedisPool[] getAllPools() {
        return Arrays.copyOf(pools, pools.length);
    }
    
    private final long hash(String shardName) {
    	return algo.hash(shardName);
    }
    
}
